package com.xiguaji.tiktok.task.aweme.event;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.Subscribe;
import com.gugee.tiktok.common.model.dal.blogger.AwemeCommentRepository;
import com.gugee.tiktok.common.model.dal.blogger.AwemeRepository;
import com.gugee.tiktok.common.model.es.ESAweme;
import com.gugee.tiktok.common.model.es.ESAwemeUpsertTmp;
import com.gugee.tiktok.common.model.mysql.Aweme;
import com.gugee.tiktok.common.model.mysql.BloggerMain;
import com.gugee.tiktok.common.model.spider.aweme.CategoriesResp;
import com.xiguaji.boot.eventbus.Event;
import com.xiguaji.boot.eventbus.IEventListener;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.tiktok.task.aweme.client.BloggerClient;
import com.xiguaji.tiktok.task.aweme.spider.AwemeSpiderClient;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class RefreshAwemeCategoriesEventListener implements IEventListener<RefreshAwemeCategoriesEventListener.RefreshAwemeCategoriesEvent> {

    @Autowired
    AwemeRepository awemeRepository;
    @Autowired
    RestHighLevelClient highLevelClient;
    @Autowired
    AwemeSpiderClient awemeSpiderClient;
    @Autowired
    AwemeCommentRepository awemeCommentRepository;
    @Autowired
    ESAweme.ESAwemeMapper esAwemeMapper;
    @Autowired
    BloggerClient bloggerClient;
    @Autowired
    AppConfig appConfig;
    long seconds7day=3600*24*7;
      long seconds3day=3600*24*3;
    @SneakyThrows
    @Override
    @Subscribe
    @AllowConcurrentEvents
    public void handle(RefreshAwemeCategoriesEvent event) {
        long lastAwemeTypeTime = (System.currentTimeMillis()/1000-appConfig.getAwemeTypeIntervalHours()*3600);
        List<Aweme> awemes =  awemeRepository.findAwemesByoUidAndoCreateTimeGreaterThan(event.bloggerUid, DateTimeUtils.second(LocalDateTime.now().minusDays(appConfig.getLatestDays())));
        // 视频小于10不分类
        if(awemes.size() < appConfig.getAwemes()){
            return;
        }
        AtomicInteger newAwemeType= new AtomicInteger();
        awemes.forEach(a -> {
            try {
                /**
                 * 视频分类依据来源 视频标签：1 模型：0.
                 * 值为1时，说明之前结果可靠，不需再分类
                 */
                if (a.getAwemeTypeSource() == 1 || a.getLastAwemeTypeTime()>lastAwemeTypeTime) {
                    return;
                }
                //标题不含标签
                boolean noTags=StringUtils.isEmpty(a.getODesc()) || !a.getODesc().contains("#");
                String comments = null;
                if(!noTags){
                    //如果标签无法分析出类别，则需依靠评论辅助
                    if (a.getSynComments() > 0 && a.getAwemeTypeComments() < a.getSynComments()) {
                        comments = getRequestBody(a.getOAwemeId());
                    }

                }else {
                    /**
                     * 添加视频分类属性，视频在三天后只做一次视频分类，因为三天后评论大概率已经够做分类分析
                     */
                    if (a.getCreateTime() > System.currentTimeMillis() - seconds3day) {
                        return;
                    }


                    /**
                     * 采集到的评论数与上一次视频分类时使用的评论数一样，则没必要再分类
                     */
                    if (a.getSynComments() > 0 && a.getAwemeTypeComments() >= a.getSynComments()) {
                        return;
                    }
                    //不含标签时，只能依赖于评论分析出分类
                    if (a.getSynComments() == 0) {
                        return;
                    }


                    comments = getRequestBody(a.getOAwemeId());

                    //获取S3评论为空且
                    if(StringUtils.isEmpty(comments) ){
                        log.error("获取oss评论为空且标题不含标签 SynComments:{}, bloggerId:{},awemeId:{} ",a.getSynComments(),a.getOUid(),a.getOAwemeId());
                        XxlJobLogger.log(" 获取oss评论为空且标题不含标签  bloggerId:{},getCategories error! awemeId:{} ",a.getOUid(),a.getOAwemeId());
                        return ;
                    }
                }

                CategoriesResp tmpType=getCategories(a,comments);
                if(!StringUtils.isEmpty(tmpType.getCatagories())){
                    a.setAwemeType(tmpType.getCatagories());
                    a.setAwemeTypeSource(tmpType.getFlag());
                    a.setLastAwemeTypeTime(System.currentTimeMillis()/1000);
//                    System.out.println("tmpType.getFlag()="+tmpType.getFlag()+":awemeid="+a.getOAwemeId());
                    if(tmpType.getFlag()==0){
                        a.setAwemeTypeComments(a.getSynComments());
                    }
                    newAwemeType.getAndIncrement();
                }
            } catch (Exception ex) {
                log.error("Get AI Categories uid={},awemeId={},{}" , a.getOUid(),a.getOAwemeId(), ex.getMessage());
                XxlJobLogger.log("Get AI Categories uid={},awemeId={},{}" , a.getOUid(),a.getOAwemeId(), ex.getMessage());
            }
        });

        if(newAwemeType.get()==0){
            return;
        }

        val bloggerMain = bloggerClient.getBloggerMainByOUid(event.bloggerUid);


        Map<String,Integer> rels = new HashMap<>();

        List<String> categories = new ArrayList<>();
        awemes.forEach(aweme -> {
            if(!StringUtils.isEmpty(aweme.getAwemeType())){
                categories.add(aweme.getAwemeType());
            }
        });

        ArrayList<Tag> list =new ArrayList<>();
        for (String a : categories) {
            if(StringUtils.isEmpty(a)){
                continue;
            }

            if(null == rels.get(a)){
                int count = Collections.frequency(categories,a);
                list.add(new Tag(a,(count+0.0)/categories.size()));
                rels.put(a,count);
            }
        }
        Collections.sort(list);
        bloggerMain.setTags(JSON.toJSONString(list));
        bloggerMain.setUpdateBy("RefreshAwemeCategoriesEvent -> categories");
        /**
         * 播主信息处理
         */
        for (Tag val : list) {
            /**
             * 如果有非other的分类
             */
            if(!val.getTypeName().equals("other")){
                /**
                 * 未审核过则更新播主分类
                 */
                if(bloggerMain.getBloggerTypeAuditStatus()!=1){
                    bloggerMain.setBloggerType(val.getTypeName());
                    bloggerMain.setBloggerTypeAuditStatus(0);
                }
                XxlJobLogger.log("uid={},BloggerType={},tags={}" , bloggerMain.getOUid(),val.getTypeName(), bloggerMain.getTags());
                break;
            }
        }

        if(!StringUtils.isEmpty(bloggerMain.getBloggerType())){
            awemes.forEach(aweme -> aweme.setBloggerType(bloggerMain.getBloggerType()));
            awemeRepository.batchUpdateType(awemes);
            saveAwemesToEs(awemes,bloggerMain);
            bloggerClient.saveBloggerMain(bloggerMain);
       }

    }
    @Setter
    @Getter
    @ToString
    class Tag implements Comparable<Tag>{
        public String typeName;
        public double percent;

        public Tag(String typeName,double percent) {
            this.typeName = typeName;
            this.percent=(double)Math.round(percent * 100000)/1000;
        }
        @Override
        public int compareTo(Tag tag) {
            return  Double.compare(tag.percent,this.percent);
        }
    }
    private CategoriesResp getCategories(Aweme aweme,String comments){

        try{
            long start = System.currentTimeMillis();
            CategoriesResp result = awemeSpiderClient.getCategories(aweme.getODesc(),comments);
            if((System.currentTimeMillis()-start)/1000 >30){
                 XxlJobLogger.log(" bloggerId:{} awemeId:{} use time:{}",aweme.getOUid(),aweme.getOAwemeId(),(System.currentTimeMillis()-start)/1000+"s");
                log.warn(" bloggerId:{} awemeId:{} use time:{}",aweme.getOUid(),aweme.getOAwemeId(),(System.currentTimeMillis()-start)/1000+"s");
            }
            return result;
        }catch (Exception e){
            log.error(" ai getCategories error!  bloggerId:{},awemeId:{} ,{}",aweme.getOUid(),aweme.getOAwemeId(),e.getMessage());
            XxlJobLogger.log(" ai  bloggerId:{},getCategories error! awemeId:{},{} ",aweme.getOUid(),aweme.getOAwemeId(),e.getMessage());

        }

        return null;

    }



    private String getRequestBody(long awemeId){
        String commentStr = "";
        long start = System.currentTimeMillis();
        try{

            String comment = "";
            val aweme = awemeRepository.findByoAwemeId(awemeId);
            if(aweme.isPresent()){
                comment = awemeCommentRepository.get(awemeId,aweme.get().getDatecode());
            }
            if((System.currentTimeMillis()-start)/1000>10)
                log.info(" get oss data awemeId:{} use time:{}s",awemeId,(System.currentTimeMillis()-start)/1000);
            List<Map<String,String>> comments = new ArrayList<>();
            comments = JsonUtil.jsonParse(comments.getClass(),comment);

            for (int i = 0; i < comments.size(); i++) {
                if(i != 0){
                    commentStr +="|";
                }
                commentStr += comments.get(i).get("text");
            }
        }catch (Exception e){
            log.error(" get oss data awemeId:{} use time:{}s,{}",awemeId,(System.currentTimeMillis()-start)/1000,e.getMessage());
        }

        return commentStr;

    }



    public static class RefreshAwemeCategoriesEvent extends Event {
        public RefreshAwemeCategoriesEvent(Long bloggerUid) {
            super( null,"RefreshAwemeCategoriesEvent");
            this.bloggerUid = bloggerUid;
        }

        public RefreshAwemeCategoriesEvent(Event event ,Long bloggerUid) {
            super( event,"RefreshAwemeCategoriesEvent");
            this.bloggerUid = bloggerUid;
        }
        
        private Long bloggerUid;
    }
    private final String AWEME_INDEX = "aweme_index_";
    @Autowired
    ESAwemeUpsertTmp.ESAwemeMapper esAwemeMapperUpsert;
    private void saveAwemesToEs(List<Aweme> awemes, BloggerMain bloggerMain){
        long start = System.currentTimeMillis();
        if(awemes.size() > 0){
            BulkRequest bulkRequest = new BulkRequest();
//            var bloggerMain = bloggerClient.getBloggerMainByOUid(bloggerUid);
            awemes.forEach( a->{
//                System.out.println(a.getOAwemeId()+":"+a.getOUid()+":"+a.getOTextExtra());
                ESAwemeUpsertTmp esAweme = esAwemeMapperUpsert.mapper(a);
                esAweme.setUniqueId(bloggerMain.getOUniqueId());
                String jsonAweme= JsonUtil.parseJson(esAweme);
                UpdateRequest updateRequest = new UpdateRequest(AWEME_INDEX+getDateStr(esAweme.getCreateTime()), String.valueOf(esAweme.getId()))
                        .doc(jsonAweme, XContentType.JSON)
                        .upsert(jsonAweme,XContentType.JSON);
                bulkRequest.add(updateRequest);
            });
            try {
                val bulk = highLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                if((System.currentTimeMillis()-start)/1000>5)
                    log.info("ESSaveAweme upsert count:{},use time={},uniqueId={}",awemes.size(),(System.currentTimeMillis()-start)/1000,bloggerMain.getOUniqueId());
            } catch (IOException e) {
                log.error("ES RefreshBlogger error",e);
            }
        }
    }
    private String getDateStr(long second){
        return DateUtil.format(DateUtil.calendar(second*1000).getTime(),"yyyyMM");
    }
}
